package com.maben.mqtt.config;

import com.maben.mqtt.constant.YmlConstant;
import com.maben.mqtt.consumer.MyConsumerHandler;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;

/**
 * 消费端配置项
 */
@Configuration
public class MqttConsumerConfig {
    @Resource
    private YmlConstant ymlConstant;

    /**
     * MQTT信息通道（生产者）
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MqttPahoClientFactory mqttConsumerFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        //设置url
        options.setServerURIs(StringUtils.split(ymlConstant.getConsumerHostUrl(), ","));
        //设置连接密码
        options.setPassword(ymlConstant.getConsumerPassword().toCharArray());
        //设置连接用户名
        options.setUserName(ymlConstant.getConsumerUsername());
        //设置超时时间，单位为秒
        options.setConnectionTimeout(ymlConstant.getConsumerConnectTimeout());
        //是否清空session，设置为false表示服务器会保留客户端的连接记录（订阅主题，qos），客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
        //设置为true表示每次连接到服务端都是以新的身份
        options.setCleanSession(ymlConstant.getConsumerCleanSession());
        //设置心跳时间 单位为秒，表示服务器每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(ymlConstant.getConsumerKeepAliveInterval());
        //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
        options.setWill("willTopic", (ymlConstant.getConsumerClientId() + "与服务器断开连接").getBytes(), 0, false);
        factory.setConnectionOptions(options);
        return factory;
    }

    //配置client,监听的topic
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(ymlConstant.getConsumerClientId(), mqttConsumerFactory());
        adapter.addTopic(ymlConstant.getConsumerDefaultTopic().split(","));
        adapter.setCompletionTimeout(5000);
        adapter.setQos(1);
        //设置接收时以byte[]接收
        final DefaultPahoMessageConverter pahoMessageConverter = new DefaultPahoMessageConverter();
        pahoMessageConverter.setPayloadAsBytes(true);
        adapter.setConverter(pahoMessageConverter);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    //通过通道获取数据
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")  //异步处理
    public MessageHandler handler() {
        return new MyConsumerHandler();
    }
}
